package com.atldy.commerce.service;

import com.alibaba.fastjson.JSON;
import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandKey;
import com.netflix.hystrix.HystrixCommandProperties;
import com.netflix.hystrix.HystrixObservableCommand;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.client.ServiceInstance;
import rx.Observable;
import rx.Subscriber;

import java.util.Collections;
import java.util.List;

/**
 * @version 1.0
 * @Author: li-dong-yang
 * @CreateTime: 2021-09-10 6:59
 * @Description： 信号量隔离
 */
@Slf4j
public class NacosClientHystrixObservableCommand extends HystrixObservableCommand<List<ServiceInstance>> {

    /**
     * <h3>需要保护的服务</h3>
     */
    private final NacosClientService nacosClientService;

    /**
     * <h3>需要传递的参数</h3>
     */
    private final List<String> serviceIds;


    public NacosClientHystrixObservableCommand(NacosClientService nacosClientService, List<String> serviceIds) {
        super(
                HystrixObservableCommand.Setter.withGroupKey(HystrixCommandGroupKey
                        .Factory.asKey("nacosClientService")) //分组名称
                        .andCommandKey(HystrixCommandKey.Factory.asKey("NacosClientHystrixObservableCommand"))
                        .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                                .withFallbackEnabled(true)          // 开启降级
                                .withCircuitBreakerEnabled(true)    // 开启熔断器
                        )
        );
        this.nacosClientService = nacosClientService;
        this.serviceIds = serviceIds;

    }

    /**
     * <h3>需要保护调用的方法</h3>
     */
    @Override
    protected Observable<List<ServiceInstance>> construct() {
        return Observable.create(subscriber -> {
            try {
                if (!subscriber.isUnsubscribed()) {
                    log.info("subscriber command task: [{}], [{}]",
                            JSON.toJSONString(serviceIds),
                            Thread.currentThread().getName());
                    serviceIds.forEach(
                            s -> subscriber
                                    .onNext(nacosClientService.getServiceInstance(s))
                    );
                    subscriber.onCompleted();
                    log.info("command task completed: [{}], [{}]",
                            JSON.toJSONString(serviceIds),
                            Thread.currentThread().getName());
                }
            } catch (Exception ex) {
                subscriber.onError(ex);
            }
        });
    }


    /**
     * <h3>兜底方法</h3>
     */
    @Override
    protected Observable<List<ServiceInstance>> resumeWithFallback() {
        return Observable.create(new Observable.OnSubscribe<List<ServiceInstance>>() {
            @Override
            public void call(Subscriber<? super List<ServiceInstance>> subscriber) {
                try {
                    if (!subscriber.isUnsubscribed()) {
                        log.info("(fallback) subscriber command task: [{}], [{}]",
                                JSON.toJSONString(serviceIds),
                                Thread.currentThread().getName());
                        subscriber.onNext(Collections.emptyList());
                        subscriber.onCompleted();
                        log.info("(fallback) command task completed: [{}], [{}]",
                                JSON.toJSONString(serviceIds),
                                Thread.currentThread().getName());
                    }
                } catch (Exception ex) {
                    subscriber.onError(ex);
                }
            }
        });
    }

}